package com.itheima.rpc.server.boot.nett;

import com.itheima.rpc.netty.codec.FrameDecoder;
import com.itheima.rpc.netty.codec.FrameEncoder;
import com.itheima.rpc.netty.codec.RpcRequestDecoder;
import com.itheima.rpc.netty.codec.RpcResponseEncoder;
import com.itheima.rpc.netty.handler.RpcRequestHandler;
import com.itheima.rpc.server.boot.RpcServer;
import com.itheima.rpc.server.config.RpcServerConfiguration;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.NettyRuntime;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.UnorderedThreadPoolEventExecutor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @desc 基于Netty实现RPC调用
 * @author:ivan.liu
 */
@Component
public class NettyRpcServer implements RpcServer {

    @Autowired
    private RpcServerConfiguration configuration;

    @Override
    public void start() {
        //创建boss，worker，business
        EventLoopGroup boss=new NioEventLoopGroup(1,new DefaultThreadFactory("boss"));
        EventLoopGroup worker=new NioEventLoopGroup(0,new DefaultThreadFactory("worker"));
        EventExecutorGroup business=new UnorderedThreadPoolEventExecutor((NettyRuntime.availableProcessors()*2)+1,new DefaultThreadFactory("business"));
        //共享业务handler
        RpcRequestHandler rpcRequestHandler = new RpcRequestHandler();
        try {
            //创建启动引导类
            ServerBootstrap serverBootstrap=new ServerBootstrap();
            serverBootstrap
                    //设置Reactor模型
                    .group(boss,worker)
                    //设置通道类型
                    .channel(NioServerSocketChannel.class)
                    //设置服务端日志
                    .handler(new LoggingHandler(LogLevel.INFO))
                    //设置服务相关参数
                    .option(ChannelOption.SO_BACKLOG,1024)
                    //设置客户端连接相关参数
                    .childOption(ChannelOption.TCP_NODELAY,true)
                    .childOption(ChannelOption.SO_KEEPALIVE,true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            //依次添加一次解码，二次解码，一次编码，二次编码，业务handler
                            pipeline.addLast("frameEncoder",new FrameEncoder());
                            pipeline.addLast("responseEncoder",new RpcResponseEncoder());
                            pipeline.addLast("frameDecoder",new FrameDecoder());
                            pipeline.addLast("requestDecoder",new RpcRequestDecoder());
                            pipeline.addLast(business,"requestHandler",rpcRequestHandler);
                        }
                    });
            //启动服务端
            ChannelFuture future = serverBootstrap.bind(configuration.getRpcPort()).sync();
            //为关闭添加监听，避免阻塞当前线程
            future.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> {
                boss.shutdownGracefully();
                worker.shutdownGracefully();
                business.shutdownGracefully();
            });
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}
